package com.zheng.cms.rocketmq.listener;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import org.apache.log4j.Logger;

import java.util.List;

/**
 * 2018-12-12 10:25:14,722 [ConsumeMessageThread_1] INFO  [com.zheng.cms.rocketmq.listener.StringMessageListener] - msg : 這是spring集成
 * <p>
 * 原来是配置没有配对
 *
 * <bean id="Consumers" class="com.zheng.cms.rocketmq.consumer.Consumers" init-method="init" destroy-method="destroy">
 * <constructor-arg name="consumerGroup" value="rocketmq-test" />
 * <constructor-arg name="namesrvAddr" value="127.0.0.1:9876" />
 * <constructor-arg name="instanceName" value="test" />
 * <constructor-arg name="topic" value="TestTopic3" />
 * <constructor-arg name="messageListener" ref="messageListener" />
 * </bean>
 * <bean id="messageListener" class="com.zheng.cms.rocketmq.listener.StringMessageListener" />
 * ，ConsumeConcurrentlyContext实现的是普通消息类型
 * 设置一个Listener，主要进行消息的逻辑处理
 * 注意这里使用的是MessageListenerConcurrently这个接口
 * consumer.registerMessageListener(new MessageListenerConcurrently() {
 *
 * @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
 * ConsumeConcurrentlyContext context) {
 * <p>
 * System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
 * <p>
 * 返回消费状态
 * CONSUME_SUCCESS 消费成功
 * RECONSUME_LATER 消费失败，需要稍后重新消费
 * return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
 */
public class StringMessageListener implements MessageListenerConcurrently {
    private Logger log = Logger.getLogger(getClass());

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            log.info("StringMessageListener,msg : " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
